package com.iwhalecloud.uncc.mq.consumer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.iwhalecloud.uncc.common.SysConstant;
import com.iwhalecloud.uncc.dao.SysConfigDao;
import com.iwhalecloud.uncc.model.MqClusterConfig;
import com.iwhalecloud.uncc.model.MqConsumerConfig;
import com.iwhalecloud.uncc.mq.listener.MqListener;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class MqConsumerManager {
	private Map<String, MqConsumer> mqConsumerMap = new ConcurrentHashMap<>();

	@Value("${mq.cluster-code}")
	private String clusterCode;
	@Value("${mq.type}")
	private String mqType;
	@Autowired
	private SysConfigDao sysConfigDao;

	public MqConsumer createMqConsumer(MqConsumerConfig mqConsumerConfig, MqListener messageListener) {
		MqClusterConfig clusterConfig = sysConfigDao.getMqClusterConfig(clusterCode);
		MqConsumer consumer = null;
		try {
			if (mqConsumerMap.get(mqConsumerConfig.getTopic()) == null) {
				if (StringUtils.isNotEmpty(mqType)) {
					if (SysConstant.CTGMQ.equalsIgnoreCase(mqType)) {
						consumer = createCtgMqPushConsumer(clusterConfig, mqConsumerConfig, messageListener);
					}
					if (SysConstant.ROCKETMQ.equalsIgnoreCase(mqType)) {
						consumer = createRocketMqConsumer(clusterConfig, mqConsumerConfig, messageListener);
					}
					System.setProperty("mqType", mqType);
				}
				if (consumer != null) {
					log.info(mqConsumerConfig.getConsumerName() + "_MqConsumer 创建完毕");
					mqConsumerMap.put(mqConsumerConfig.getTopic(), consumer);
				}
			}
		} catch (Exception e) {
			log.error(mqConsumerConfig.getConsumerName() + "_MqConsumer 创建异常" + ExceptionUtils.getStackTrace(e));
		}
		return consumer;
	}

	private CtgMqPushConsumer createCtgMqPushConsumer(MqClusterConfig clusterConfig, MqConsumerConfig mqConsumerConfig,
			MqListener messageListener) throws Exception {
		CtgMqPushConsumer consumer = new CtgMqPushConsumer(clusterConfig, mqConsumerConfig, messageListener);
		consumer.init();
		consumer.listen();
		return consumer;
	}

	private RocketMqConsumer createRocketMqConsumer(MqClusterConfig clusterConfig, MqConsumerConfig mqConsumerConfig,
			MqListener messageListener) throws Exception {
		RocketMqConsumer consumer = new RocketMqConsumer(clusterConfig, mqConsumerConfig, messageListener);
		consumer.init();
		consumer.listen();
		return consumer;
	}

	public MqConsumer getMqConsumerByConsumerName(String consumerName) {
		return mqConsumerMap.get(consumerName);
	}

}
